package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import redis.clients.jedis.JedisCluster;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Map;

/**
 * @author : GW00280745
 * @date : 2022/8/15 9:39
 * @description :
 */
public class MapBatterAvg implements Serializable {
    private MapStateDescriptor<String, Map<String, Object>> confInfo = new MapStateDescriptor<>("conf-info", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(String.class, Object.class));
    private String redisNameSpaceBatter = ChargePredictFields.ins().redisNameSpaceBatter;
    public MapBatterAvg(){
    }
    public void process(JSONObject line,
                        MapState<String, String> latestMoveAvgState,
                        KeyedBroadcastProcessFunction.ReadOnlyContext ctx,
                        JedisCluster jedis,
                        MapState<String, Map> mapTempAvgState){
        try {
            avg(line,latestMoveAvgState,ctx,jedis,mapTempAvgState);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void avg(JSONObject line,
                     MapState<String, String> latestMoveAvgState,
                     KeyedBroadcastProcessFunction.ReadOnlyContext ctx,
                     JedisCluster jedis,
                     MapState<String, Map> mapTempAvgState){
        try {
            ReadOnlyBroadcastState<String, Map<String, Object>> broadcastState = ctx.getBroadcastState(confInfo);
            Map<String, Object> confMap = broadcastState.get(null);
            //校验是否为间隔点
            String intervalPoint = line.getString(ChargePredictFields.ins().intervalPointFieldBatter);
            if (StringUtils.isBlank(intervalPoint)){
//                String latestMoveAvg = latestMoveAvgState.get("latest")
//                line.put(StdFieldsConstant.stdField_batterMapValue,latestMoveAvg)
                return;
            }
            String vinVal = line.getString(ChargePredictFields.ins().srcFieldVin);
            Long tidVal = line.getLong(ChargePredictFields.ins().srcFieldTid);
            BigDecimal cap = line.getBigDecimal(ChargePredictFields.ins().srcFieldBatterCapacity);
            //1.获取当前记录得电流
            BigDecimal batter = line.getBigDecimal(ChargePredictFields.ins().dwsFieldBatter);
            BigDecimal bmsSoc = line.getBigDecimal(ChargePredictFields.ins().srcFieldBmsSoc);
            //2.获取map表中得移动平均值，需要和redis交互 10.255.128.239:6370
            String redisKey = AxisInfo.singleInstance().axisBatter(line,null,null,confMap,true);
            redisKey = redisNameSpaceBatter.concat(redisKey);
            String axisBatterDefault = ChargeInfo.single().axisBatterDefault(line,null);
            String redisKeyDefault = redisNameSpaceBatter.concat(axisBatterDefault);
            String mapValJson = ChargeInfo.single().getRedisHvalNoDefault(jedis,redisKey,vinVal);
//            String map_val_jsonDefault = ChargeUtil.single().getRedisHVal(jedis,redisKeyDefault,EndTempConstant.redisDefaultField)
            String mapValJsonDefault = ChargeInfo.single().getRedisHvalNoDefault(jedis,redisKeyDefault,vinVal);
            JSONObject mapVal = StringUtils.isBlank(mapValJson)?new JSONObject():JSONObject.parseObject(mapValJson);
            JSONObject mapValDefault = StringUtils.isBlank(mapValJsonDefault)?new JSONObject():JSONObject.parseObject(mapValJsonDefault);
            //3.计算最新得移动平均值
            String currentVal = batter.toString().concat(ChargePredictFields.ins().char3).concat(tidVal.toString());
            String moveAvg = moveAvg(mapVal,currentVal,line);
            String moveAvgDefault = moveAvg(mapValDefault,currentVal,line);
            //4.更新到redis
            jedis.hset(redisKey,vinVal,moveAvg);
//            jedis.hset(redisKeyDefault,EndTempConstant.redisDefaultField,moveAvgDefault)
            jedis.hset(redisKeyDefault,vinVal,moveAvgDefault);
//            LogTE.info(String.format("电流表 #&# statebackend #&# 查询&更新 #&# key=%s #&# field=%s #&# 查询value=%s #&# 更新value=%s",redisKey,vinVal,map_val_json,moveAvg))
            if (bmsSoc.compareTo(new BigDecimal(num94))>=0 ){
                if (new BigDecimal(num156Str).compareTo(cap)!=0){
                    //soc快满时电流降低，但温度不一定预测准确 而且 对400车型来说电流基本不可信
                    String axisSoc = AxisInfo.singleInstance().axis(bmsSoc.toString(), 94, 100, 6);
                    String redisKeySoc = redisNameSpaceBatter.concat(axisSoc);
                    String mapValSoc1 = jedis.hget(redisKeySoc, vinVal);
                    JSONObject mapValSoc = StringUtils.isBlank(mapValSoc1)?new JSONObject():JSONObject.parseObject(mapValSoc1);
                    String moveAvgSoc = moveAvg(mapValSoc,currentVal,line);
                    jedis.hset(redisKeySoc,vinVal,moveAvgSoc);
                    avgVal2State(mapTempAvgState,vinVal,redisKeySoc,moveAvgSoc);
                }else {
                    //400的车型
                    String chargeGroup = line.getString(ChargePredictFields.ins().dwsFieldChargeBegin);
                    String s = latestMoveAvgState.get(chargeGroup);
                    if (StringUtils.isBlank(s)){
                        String tidSoc = tidVal.toString().concat(ChargePredictFields.ins().char3).concat(bmsSoc.toString());
                        latestMoveAvgState.put(chargeGroup,tidSoc);
                    }else {
                        double l = Double.parseDouble(bmsSoc.toString());
                        if (l >= num99){
                            String[] split = s.split(ChargePredictFields.ins().char3);
                            Long tidSocTid = Long.parseLong(split[0]);
                            BigDecimal tidSocSoc = new BigDecimal(split[1]);
                            //5 * 36 * 1000 / (item_time_end - item_time_95)
                            BigDecimal factor = bmsSoc.subtract(tidSocSoc)
                                    .multiply(new BigDecimal(36 * 1000))
                                    .divide(new BigDecimal(tidVal - tidSocTid), BigDecimal.ROUND_HALF_UP);
                            String axisSoc = AxisInfo.singleInstance().axis(bmsSoc.toString(), 94, 100, 6);
                            String redisKeySoc = ChargePredictFields.ins().redisNameSpaceBatterSoc.concat(axisSoc);
                            String mapValSoc1 = jedis.hget(redisKeySoc, vinVal);
                            JSONObject mapValSoc = StringUtils.isBlank(mapValSoc1)?new JSONObject():JSONObject.parseObject(mapValSoc1);
                            String moveAvgSoc = AxisInfo.singleInstance().batterSocAvg(factor,tidVal,mapValSoc).toJSONString();
                            jedis.hset(redisKeySoc,vinVal,moveAvgSoc);
                            avgVal2State(mapTempAvgState,vinVal,redisKeySoc,moveAvgSoc);
                        }
                    }
                }
            }
            //4.1 更新到state
            avgVal2State(mapTempAvgState,vinVal,redisKey,moveAvg);
            avgVal2State(mapTempAvgState,vinVal,redisKeyDefault,moveAvgDefault);
            //5.更新数据
            String batterMoveAvg = JSONObject.parseObject(moveAvg).getString("batterMoveAvg");
            latestMoveAvgState.put("latest",batterMoveAvg);
            line.put(ChargePredictFields.ins().stdFieldBatterMapValue,batterMoveAvg);
            line.put(ChargePredictFields.ins().stdFieldBatterValueErr,ChargePredictFields.ins().zeroStr);
            //end
        }catch (Exception e){
//            LogTE.info(ExceptionUtils.getFullStackTrace(e))
            e.printStackTrace();
            line.put("batter_map_phase",ChargePredictFields.ins().err);
            line.put(ChargePredictFields.ins().stdFieldBatterValueErr,ChargePredictFields.ins().oneStr);
        }
    }
    private int num94=94;
    private int num99=99;
    private String num156Str="156";
    /**
     * @description: 移动平均值计算方式 : ( map_table_val + val1 + val2 ) / 3
     * @author:  GW00280745
     * @date: 2022/5/9 10:46
     * @param: mapVal  map表中得数据格式:{move_avg:val,tid1:temp_acc_val,tid2:temp_acc_val}
     * @param: val vin#tid#acc
     * @return: java.lang.String
     **/
    private String moveAvg(JSONObject mapVal,String val,JSONObject dwsLine){
        String startsWith = ChargePredictFields.ins().batterMoveAvgField;
        String pointField = ChargePredictFields.ins().dwsFieldBatterMapValue;
        String errField= ChargePredictFields.ins().dwsFieldBatterValueErr;
        String avgField = ChargePredictFields.ins().batterMoveAvgField;
        return ChargeInfo.single().moveAvg(mapVal,val,dwsLine,startsWith,pointField,errField,avgField);
    }

    private void avgVal2State(MapState<String, Map> latestMoveAvgState, String vinVal, String redisKey, String moveAvg) throws Exception {
        Map map = latestMoveAvgState.get(redisKey);
        if (map==null){
            map = new JSONObject();
        }
        map.put(vinVal,moveAvg);
        latestMoveAvgState.put(redisKey,map);
    }
}
